/**
 * This file is a part of qloudgen-sec. 
 * You can redistribute qloudgen-sec and/or modify it under the terms of the Lesser GNU General Public License version 3. 
 * qloudgen-sec is distributed WITHOUT ANY WARRANTY. 
 *
 * See the Lesser GNU General Public License for more details at http://www.gnu.org/licenses/. 
 **/
package org.redisson.cluster;

import java.net.URI;
import java.net.URISyntaxException;

import java.util.List;
import java.util.ArrayList;
import java.util.Queue;
import java.util.Map;
import java.util.HashMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.lambdaworks.redis.RedisClient;
import com.lambdaworks.redis.RedisConnection;
import com.lambdaworks.redis.RedisException;
import com.lambdaworks.redis.codec.RedisCodec;
//import com.lambdaworks.redis.pubsub.RedisPubSubAdapter;
//import com.lambdaworks.redis.pubsub.RedisPubSubConnection;

import org.redisson.Config;
import org.redisson.codec.RedisCodecWrapper;

/**
 * <p>Title: Redis cluster connections support</p>
 * <p>Description: Wrapper redisson connection manager cluster implementation.</p>
 * <p>Copyright: qloudgen/sec 2014</p>
 * @author <a href="mailto:hyq.dd.sun@gmail.com">glue.3M</a>
 */
public class ClusterConnectionManager {
	/**
	 * Log output.
	 */
	private final static Logger logger = LoggerFactory.getLogger( ClusterConnectionManager.class );
	/**
	 *
	 */
	private final Map<URI, NodePool> pools = new ConcurrentHashMap<URI, NodePool>();
	/**
	 *
	 */
	private final Map<Integer, NodePool> slots = new ConcurrentHashMap<Integer, NodePool>();
	/**
	 *
	 */
	private final Semaphore activeConnections;
	/**
	 *
	 */
	private final RedisCodec codec;
	/**
	 *
	 */
	private final String password;
	/**
	 *
	 */
	private final int connectionPoolSize;
	/**
	 *
	 */
	private final Config config;
	
	/**
	 * Custom constructor.
	 *
	 * @param config
	 */
	public ClusterConnectionManager(Config config) {
		/**
		 * Initialize common parameters.
		 */
		this.connectionPoolSize = config.getConnectionPoolSize();
		this.password = config.getPassword();
		this.codec = new RedisCodecWrapper( config.getCodec() );
		this.activeConnections = new Semaphore( connectionPoolSize );
		this.config = config;
		/**
		 * Initialize all cluster information from address list.
		 */
		discoverClusterNodesAndSlots( config );
	}
	
	/**
	 * Discover all cluster nodes and slots.
	 *
	 * @param config
	 */
	private void discoverClusterNodesAndSlots(Config config) {
		// Loop address.
		for ( URI address : config.getAddresses() ) {
			try {
				discoverClusterNodes( address );
				break;
			}
			catch (RedisException e) {
				logger.error( "Current find redis cluster error and will using next address...\n{}" , e );
			}
		}
	}
	
	/**
	 * Returns current nodes size.
	 *
	 * @return int
	 */
	public int size() {
		int s = 0;
		for ( NodePool pool : pools.values() ) {
			if ( pool.getNum() >= 0 ) {
				s = s + 1;
			}
		}
		return s;
	}
	
	/**
	 * Shutdown operation.
	 */
	public void shutdown() {
		for ( NodePool pool : pools.values() ) {
			shutdown( pool );
		}
		/**
		 * Remove all pool.
		 */
		pools.clear();
		/**
		 * Remove all slot.
		 */
		slots.clear();
		/**
		 * Release all lock.
		 */
		activeConnections.release( connectionPoolSize );
	}
	
	/**
	 * Returns current address.
	 *
	 * @param key
	 * @return URI
	 */
	public URI getAddress(Object key) {
		return getPool( key ).getAddress();
	}
	
	/**
	 * Acquire connection.
	 *
	 * @param key
	 * @return RedisConnection<K, V>
	 */
	public <K, V> RedisConnection<K, V> getConnection(Object key) {
		/**
		 * Acquire lock.
		 */
		acquireConnection();
		//
		NodePool pool = getPool( key );
		//
		logger.trace( "Current acquired redis pool is...{}" , pool );
		//
		return pool.getConnection();
	}
	
	/**
	 * Release connection.
	 *
	 * @param key
	 * @param connection
	 */
	public <K, V> void release(Object key , RedisConnection<K, V> сonnection) {
		//
		if ( сonnection != null ) {
			//
			NodePool pool = getPool( key );
			//
			logger.trace( "Current released redis pool is...{}" , pool );
			//
			pool.addConnection( сonnection );
		}
		/**
		 * Release lock.
		 */
		activeConnections.release();
	}
	
	/**
	 * Refresh by ask & moved.
	 *
	 * @param message
	 */
	public void refresh(String message) {
		/**
		 * Parse exception node message.
		 */
		String[] node = message.split( " " );
		String slot = node[1];
		String addr = node[2];
		URI currentAddress = null;
		try {
			currentAddress = new URI( "//" + addr );
		}
		catch (URISyntaxException e) {
			throw new RuntimeException( e );
		}
		/**
		 * Will update pools & slots.
		 */
		if ( ! pools.containsKey( currentAddress ) ) {
			synchronized( this ) {
				// Try again !
				if ( ! pools.containsKey( currentAddress ) ) {
					logger.debug( "Current refresh moved redis cluster pools using new address...{}" , currentAddress );
					try {
						discoverClusterNodes( currentAddress );
					}
					catch (Throwable t) {
						logger.error( "Current refresh redis cluster pools error...\n{}" , t );
					}
				}
			}
		}
		/**
		 * Only update slots.
		 * So pools contains slots will not synchronized with slots.
		 */
		NodePool pool = pools.get( currentAddress );
		NodePool old = slots.put( Integer.valueOf( slot ) , pool );
		//
		logger.info( "Current changed redis cluster slots to pools is...{}-->{}-->{}" , slot , old , pool );
	}
	/**
	 * Refresh by timeout.
	 *
	 * connection.info( "replication" ).indexOf( "role:master" ) > -1
	 *
	 * @param key
	 * @param address
	 */
	public void refresh(Object key , URI address) {
		//
		logger.debug( "Current refresh redis cluster pools using old address...{}" , address );
		NodePool old1 = getPool( key );
		/**
		 * If ( old address == current address ) means never refreshed.
		 */
		if ( old1.getAddress().equals( address ) ) {
			synchronized( this ) {
				// Try again !!!
				NodePool old2 = getPool( key );
				if ( old2.getAddress().equals( address ) ) {
					/**
					 * Must loop all available pool from slots not pools.
					 *
					 * 1)name = master
					 * 2)status = ok
					 * 3)has slots values
					 */
					for ( NodePool tmp : slots.values() ) {
						/**
						 * Find another available redis connection to rediscover.
						 * Must not current address pool.
						 */
						if ( ! tmp.getAddress().equals( address ) ) {
							logger.debug( "Current refresh timeout redis cluster pools using new address...{}" , tmp.getAddress() );
							//
							try {
								discoverClusterNodes( tmp.getAddress() );
								return;
							}
							catch (Throwable t) {
								logger.error( "Current refresh redis cluster pools error...\n{}" , t );
							}
						}
					}
					//
					logger.info( "Current refreshed redis cluster pools never successed..." );
					return;
				}
			}
		}
		//
		logger.info( "Current refreshed redis cluster pools already finished..." );
	}
	
	/**
	 * Calc key and pool.
	 *
	 * @param key
	 * @return NodePool
	 */
	private NodePool getPool(Object key) {
		//
		if ( key.getClass().equals( Integer.class ) ) {
			return getPoolFromNum( ( ( Integer )key ).intValue() );
		}
		//
		if ( key.getClass().equals( String.class ) ) {
			return getPoolFromSlot( ( String )key );
		}
		//
		throw new RuntimeException( "Unsupport key class type." );
	}
	
	/**
	 * Calc num and pool.
	 *
	 * @param num
	 * @return NodePool
	 */
	private NodePool getPoolFromNum(int num) {
		//
		logger.trace( "Current redis cluster num is...{}" , num );
		//
		for ( NodePool pool : pools.values() ) {
			if ( pool.getNum() == num ) {
				return pool;
			}
		}
		//
		return null;
	}
	
	/**
	 * Calc slot and pool.
	 *
	 * @param name
	 * @return NodePool
	 */
	private NodePool getPoolFromSlot(String name) {
		/**
		 * When using SerializationCodec set jvm charset=GBK will success!!!
		 * -Dfile.encoding=GBK
		 *
		 * Ubuntu default charset is UTF-8 and will fail.
		 */
		//logger.trace( "Current platform charset is...{}" , System.getProperty( "file.encoding" ) );
		Integer slot = ClusterCRC16.getSlot( codec.encodeKey( name ) );
		//
		logger.trace( "Current redis cluster key to slot is...{}-->{}" , name , slot );
		//
		return slots.get( slot );
	}
	
	/**
	 * Shutdown operation.
	 *
	 * @param pool
	 */
	private void shutdown(NodePool pool) {
		logger.info( "Current shutdown redis cluster pool is...{}" , pool );
		try {
			pool.shutdown();
		}
		catch (Throwable t) {
			logger.error( "Current shutdown redis cluster pool error...\n{}" , t );
		}
		//finally {
		//	pool = null;
		//}
	}
	
	/**
	 * Discover cluster nodes.
	 *
	 * 1)master --> master,fail    [phase-1][error]
	 *   slave  --> slave
	 * 2)master --> master,fail    [phase-2][ok]
	 *   slave  --> master
	 * 3)master --> slave          [phase-3][ok][after start master again]
	 *   slave  --> master
	 *
	 * @param address
	 */
	private void discoverClusterNodes(URI address) {
		//
		logger.info( "Current redis cluster address is...{}" , address );
		//
		RedisConnection<String, String> connection = null;
		try {
			/**
			 * Creates redis connection.
			 */
			RedisClient client = new RedisClient( address.getHost() , address.getPort() );
			connection = client.connect();
			/**
			 * Returns redis cluster infos and nodes.
			 */
			String infos = connection.clusterInfos();
			logger.info( "Current redis cluster status...\n{}" , infos );
			String nodes = connection.clusterNodes();
			logger.info( "Current redis cluster nodes...\n{}" , nodes );
			/**
			 * Loop all cluster nodes.
			 */
			String[] result = nodes.split( "\n" );
			for ( int i = 0 ; i < result.length ; i++ ) {
//			for ( String node : nodes.split( "\n" ) ) {
				int num = i;
				String node = result[i];
				//
				String id = null;
				String name = null;
				String status = null;
				String parent = null;
				RedisClient currentClient = null;
				URI currentAddress = null;
				List<String> currentSlot = new ArrayList<String>();
				if ( node.indexOf( "handshake" ) < 0 ) {
					/**
					 * Parse cluster node message.
					 */
					String[] strs = node.split( " " );
					//
					id = strs[0];
					//
					if ( ":0".equals( strs[1] ) ) {
						currentClient = client;
						currentAddress = address;
					}
					else {
						try {
							currentAddress = new URI( "//" + strs[1] );
						}
						catch (URISyntaxException e) {
							throw new RuntimeException( e );
						}
						currentClient = new RedisClient( currentAddress.getHost() , currentAddress.getPort() );
					}
					//
					if ( strs[2].contains( "," ) ) {
						String[] tmp = strs[2].split( "," );
						if ( "myself".equals( tmp[0] ) ) {
							name = tmp[1];
							status = "ok";
						}
						else {
							name = tmp[0];
							status = tmp[1];
						}
					}
					else {
						name = strs[2];
						status = "ok";
					}
					//
					if ( "slave".equals( name ) ) {
						num = -1; // Only add num for master nodes.
						parent = strs[3];
					}
					//
					for ( int j = 8 ; j < strs.length ; j++ ) {
						currentSlot.add( strs[j] );
					}
					/**
					 * Create node pool.
					 */
					NodePool pool = new NodePool( id , num , currentAddress , currentClient );
					pool.setName( name );
					pool.setStatus( status );
					pool.setParent( parent );
					pool.setSlots( currentSlot );
					//
					logger.info( "Current redis cluster pool is...{}" , pool );
					/**
					 * Add node pool.
					 */
					NodePool old = pools.put( currentAddress , pool );
					/**
					 * Add slot pool.
					 */
					discoverClusterSlots( pool );
					/**
					 * Shutdown old node pool.
					 */
					if ( old != null ) {
						shutdown( old );
					}
				}
			}
		}
		finally {
			/**
			 * Close current connection.
			 */
			if ( connection != null ) {
				try {
					connection.close();
				}
				catch (Throwable t) {
					logger.error( "Current close redis connection error...\n{}" , t );
				}
			}
		}
		//
		logger.info( "Current redis cluster pools...\n{}" , dump() );
	}
	
	/**
	 * Discover cluster slots.
	 *
	 * @param pool
	 */
	private void discoverClusterSlots(NodePool pool) {
		//
		if ( pool.isMaster() && ! pool.isFail() ) {
			for ( String slot : pool.getSlots() ) {
				if ( slot.contains( "-" ) ) {
					String[] slotRange = slot.split( "-" );
					for ( int i = Integer.valueOf( slotRange[0] ); i <= Integer.valueOf( slotRange[1] ); i++ ) {
						slots.put( i , pool );
					}
				}
				else {
					slots.put( Integer.valueOf( slot ) , pool );
				}
			}
		}
		//
		logger.info( "Current redis cluster slots...{}" , slots.size() );
	}
	
	/**
	 * Dump current all pools.
	 *
	 * @return String
	 */
	private String dump() {
		StringBuilder strs = new StringBuilder();
		for ( NodePool pool : pools.values() ) {
			strs.append( pool ).append( "\n" );
		}
		return strs.toString();
	}
	
	/**
	 * Lock operation.
	 */
	private void acquireConnection() {
		//
		if ( ! activeConnections.tryAcquire() ) {
			//
			logger.warn( "Connection pool gets exhausted! Trying to acquire connection ..." );
			// Wait time.
			long time = System.currentTimeMillis();
			activeConnections.acquireUninterruptibly();
			long endTime = System.currentTimeMillis() - time;
			//
			logger.warn( "Connection acquired, time spended: {} ms" , endTime );
		}
	}
	
	/**
	 *
	 */
	private class NodePool {
		/**
		 *
		 */
		private final String id;
		/**
		 *
		 */
		private final int num;
		/**
		 *
		 */
		private final URI address;
		/**
		 *
		 */
		private RedisClient client;
		/**
		 *
		 */
		private Queue<RedisConnection> connections = new ConcurrentLinkedQueue<RedisConnection>();
		/**
		 *
		 */
		private String parent;
		/**
		 *
		 */
		private String name = null;
		/**
		 *
		 */
		private String status = null;
		/**
		 *
		 */
		private List<String> slots = null;
		/**
		 *
		 */
		private boolean closed = false;
		
		/**
		 * Custom constructor.
		 *
		 * @param id
		 * @param address
		 * @param client
		 */
		public NodePool(String id , int num , URI address , RedisClient client) {
			this.id = id;
			this.num = num;
			this.address = address;
			this.client = client;
		}
		
		public String getId() {
			return id;
		}
		
		public int getNum() {
			return num;
		}
		
		public URI getAddress() {
			return address;
		}
		
		public void setParent(String parent) {
			this.parent = parent;
		}
		
		public String getParent() {
			return parent;
		}
		
		public void setName(String name) {
			this.name = name;
		}
		
		public String getName() {
			return name;
		}
		
		public void setStatus(String status) {
			this.status = status;
		}
		
		public String getStatus() {
			return status;
		}
		
		public void setSlots(List<String> slots) {
			this.slots = slots;
		}
		
		public List<String> getSlots() {
			return slots;
		}
		
		public boolean isMaster() {
			return "master".equals( name );
		}
		
		public boolean isFail() {
			return "fail".equals( status );
		}
		
		/**
		 * Returns current node connection.
		 *
		 * @return RedisConnection<K, V>
		 */
		public <K, V> RedisConnection<K, V> getConnection() {
			if ( closed ) {
				throw new RedisException( "Connection is closed." );
			}
			RedisConnection<K, V> connection = connections.poll();
			if ( connection == null ) {
				connection = client.connect( codec );
				if ( password != null ) {
					connection.auth( password );
				}
			}
			//
			logger.trace( "Current acquired redis connection is...{}" , connection );
			return connection;
		}
		
		/**
		 * Adds current node connection.
		 *
		 * @param connection
		 */
		public <K, V> void addConnection(RedisConnection<K, V> connection) {
			logger.trace( "Current released redis connection is...{}" , connection );
			connections.add( connection );
		}
		
		/**
		 * Shutdown current client.
		 */
		public void shutdown() {
			closed = true;
			try {
				client.shutdown();
			}
			finally {
				connections.clear();
				slots.clear();
			}
		}
		
		/**
		 * Returns the String representation.
		 *
		 * @return String
		 */
		@Override
		public String toString() {
			StringBuilder builder = new StringBuilder();
			builder.append( "[" );
			builder.append( super.toString() );
			builder.append( "-->" );
			builder.append( "id=" ).append( id ).append( ";" );
			builder.append( "num=" ).append( num ).append( ";" );
			builder.append( "address=" ).append( address ).append( ";" );
			builder.append( "parent=" ).append( parent ).append( ";" );
			builder.append( "name=" ).append( name ).append( ";" );
			builder.append( "status=" ).append( status ).append( ";" );
			builder.append( "slots=" ).append( slots ).append( ";" );
			builder.append( "client=" ).append( client ).append( ";" );
			builder.append( "connections=" ).append( connections.size() );
			builder.append( "]" );
			//
			return builder.toString();
		}
	}
}